package org.goldenport.g3

import scala.collection.mutable.ArrayBuffer
import scala.xml.Node
import com.asamioffice.goldenport.util.OptionUtil.lift
import org.goldenport.g3.pipes._
import org.goldenport.g3.messages.{Ping, RecordSchema, Html}
import java.lang.reflect.Method
import org.goldenport.application.GApplicationDescriptor

/**
 * @since   Mar. 19, 2010
 * @version Oct. 18, 2010
 * @version Nov. 29, 2010
 * @author  ASAMI, Tomoharu
 */
abstract class G3Pipe(aContext: G3Context) extends G3Actionable {
  private var next_pipe: Option[G3Pipe] = None
  protected var context_channel: Option[G3Channel] = None
  private var config_value: Option[AnyRef] = None

  setContext(aContext)

  private def push_pipe[T <: G3Pipe](pipe: T): T = {
    pipe.next_pipe = next_pipe
    pipe.context_channel = context_channel
    next_pipe = Some(pipe)
    pipe
  }

  protected def pipe_execute(parcel: G3Parcel) {
    pipe_next(execute_action(parcel))
  }

  protected final def pipe_next(parcel: G3Parcel) {
    next_pipe match {
      case Some(next) => next.pipe_execute(parcel)
      case None => context.sendTerminal(parcel, context_channel.get)
    }
  }

  protected final def is_processing(uow: G3UnitofworkId): Boolean = {
    if (is_Processing(uow)) true
    else if (next_pipe.isDefined) next_pipe.get.is_processing(uow)
    else false
  }

  protected def is_Processing(uow: G3UnitofworkId): Boolean = false

  protected final def get_config_value: Option[AnyRef] = config_value

  protected def parcelTitle = "pipe(" + getClass.getSimpleName + ")"

/*
  //
  protected def doAction(parcel: G3Parcel): G3Parcel = {
//    println("G3Pipe.doAction1(" + getClass + "):" + parcel.content) // XXX
    val (content, attachments) = doAction(parcel.content, parcel.attachments, parcel)
    if ((content == parcel.content) && (attachments == parcel.attachments)) {
      parcel
    } else {
      parcel.context.createParcel("pipe(" + getClass + ")")(content, attachments, parcel)
    }
  }

  protected def doAction(content: AnyRef, attachments: List[AnyRef], parcel: G3Parcel): (AnyRef, List[AnyRef]) = {
//    println("G3Pipe.doAction2(" + getClass + "):" + content + "/" + attachments) // XXX
    (doAction(content), attachments)
  }

  protected def doAction(content: AnyRef): AnyRef = {
//    println("G3Pipe.doAction3(" + getClass + "):" + content) // XXX
    content
  }
*/

  //
  def apply(conf: => AnyRef) = {
    CONFIG(conf)
  }

  def apply() = {
//    CONFIG(Ping())
    this
  }

  def CONFIG(proc: => AnyRef) = {
    context.pushPipe(this)
    config_value = Some(proc)
    context.popPipe()
    this
  }

  //
  def agent(activity: PartialFunction[AnyRef, AnyRef]): G3Agent = {
    push_pipe(new G3Agent(activity, context))
  }

  def agentc(activity: (G3AgentContext) => PartialFunction[AnyRef, AnyRef]): G3Agent2 = {
    push_pipe(new G3Agent2(activity, context))
  }

  def agentv(activity: PartialFunction[Any, Any]): G3Agent = {
    val act = activity andThen {
      case v: Boolean => v.asInstanceOf[AnyRef]
      case v: Char => v.asInstanceOf[AnyRef]
      case v: Short => v.asInstanceOf[AnyRef]
      case v: Int => v.asInstanceOf[AnyRef]
      case v: Long => v.asInstanceOf[AnyRef]
      case v: Float => v.asInstanceOf[AnyRef]
      case v: Double => v.asInstanceOf[AnyRef]
      case r: AnyRef => r
    }
    push_pipe(new G3Agent(act, context))
  }

  def agents(activity: => AnyRef): G3AgentStart = {
    push_pipe(new G3AgentStart(() => activity, context))
  }

  def agentpf(activity: PartialFunction[AnyRef, AnyRef]) = agent(activity)

  def agentf[A, B](activity: A => B): G3AgentFunction = {
    push_pipe(new G3AgentFunction(activity, context))
  }

  def agentf[A, B](activity: { def apply[A](a: A): B }): G3AgentFunction = {
    push_pipe(new G3AgentFunction(activity.apply _, context))
  }

  def agento(activity: AnyRef): G3AgentObject = {
    push_pipe(new G3AgentObject(activity, context))
  }

  def agento(activity: Class[_]): G3AgentClass = {
    push_pipe(new G3AgentClass(activity, context))
  }

  def agento(activity: String): G3AgentClass = {
    val ld = getClass.getClassLoader // XXX container class loader
    push_pipe(new G3AgentClass(ld.loadClass(activity), context)) 
  }

  def agentx(activities: AnyRef*): G3AgentOr = {
    push_pipe(new G3AgentOr(context, activities: _*))
  }

  def filter(activity: AnyRef => Boolean): G3Filter = {
    push_pipe(new G3Filter(activity, context))
  }

  def map(activity: AnyRef => AnyRef): G3Map = {
    push_pipe(new G3Map(activity, context))
  }

  // XXX channel name should be specified by WPath
  def invoke(channel: Symbol): G3Invoke = {
    invoke(channel.name)
  }

  def invoke(channel: String): G3Invoke = {
    push_pipe(new G3Invoke(channel, context))
  }

  def publish(subscribers: String*): G3Publish = {
    publish(subscribers.toList)
  }

  def publish(subscribers: List[_]): G3Publish = {
    val subs = subscribers.map {
      case s: Symbol => s.name
      case s => s.toString
    }
    push_pipe(new G3Publish(subs, context))
  }

  def join(): G3Join[AnyRef] = {
    push_pipe(new G3Join[AnyRef](context))
  }

  def split(): G3Split[AnyRef] = {
    split(NullSplitStrategy)
  }

  def split(strategy: SplitStrategy): G3Split[AnyRef] = {
    push_pipe(new G3Split(strategy, context))
  }

  def aggregate(): G3Aggregate[AnyRef] = {
    push_pipe(new G3Aggregate(context))
  }

  def transform(): G3Transform[AnyRef] = {
    push_pipe(new G3Transform(context))
  }

  def convert(rule: String): G3Convert[AnyRef] = {
    push_pipe(new G3Convert(rule, context))
  }

  def get(url: String): G3Get = {
    push_pipe(new G3Get(url, context))
  }

  def put(url: String): G3Put = {
    push_pipe(new G3Put(url, context))
  }

  def post(url: String): G3Post = {
    push_pipe(new G3Post(url, context))
  }

  def delete(url: String): G3Delete = {
    push_pipe(new G3Delete(url, context))
  }

  def mkcol(url: String): G3Mkcol = {
    push_pipe(new G3Mkcol(url, context))
  }

  def copy(from: String, to: String): G3Copy = {
    push_pipe(new G3Copy(from, to, context))
  }

  def move(from: String, to: String): G3Move = {
    push_pipe(new G3Move(from, to, context))
  }

  def html(body: => Node, title: String = "", schema: RecordSchema = null): G3Html = {
    push_pipe(new G3Html(new Html(body, title = title, schema = lift(schema)), context))
  }
}

class SplitStrategy() extends NotNull
object NullSplitStrategy extends SplitStrategy

class G3Agent(val activity: PartialFunction[AnyRef, AnyRef], context: G3Context) extends G3Pipe(context) with ActivityActionable {
  protected def create_Activity(ctx: G3AgentContext) = activity
}

class G3Agent2(val activity: G3AgentContext => PartialFunction[AnyRef, AnyRef], context: G3Context) extends G3Pipe(context) with ActivityActionable {
  protected def create_Activity(ctx: G3AgentContext) = activity(ctx)
}

class G3AgentStart(val activity: Function0[AnyRef], context: G3Context) extends G3Pipe(context) with ActivityActionable {
  protected def create_Activity(ctx: G3AgentContext) = {
    case _ => activity()
  }
}

// 2011-11-26
class G3AgentFunction(val activity: Function1[_, _], context: G3Context) extends G3Pipe(context) with ActivityActionable {
  protected def create_Activity(ctx: G3AgentContext) = new PartialFunction[AnyRef, AnyRef] {
    override def isDefinedAt(in: AnyRef): Boolean = {
      _get_method(in).isDefined
    }

    private def _get_method(in: AnyRef): Option[Method] = {
      val methods = activity.getClass().getMethods()
      val method: Option[Method] = methods find { m =>
        "apply".equals(m.getName()) && _is_valid(m, in)
      } 
      method
    }

    private def _is_valid(m: Method, in: AnyRef): Boolean = {
      val pts = m.getParameterTypes()
      pts.length == 1 &&
      pts(0).isAssignableFrom(in.getClass)
    }

    override def apply(in: AnyRef): AnyRef = {
      val method = _get_method(in).get
      method.invoke(in)
    }
  }
}

// 2011-11-26
class G3AgentObject(val activity: AnyRef, context: G3Context) extends G3Pipe(context) with ActivityActionable {
  protected def create_Activity(ctx: G3AgentContext) = new PartialFunction[AnyRef, AnyRef] {
    import UAgent._

    override def isDefinedAt(in: AnyRef): Boolean = {
        findMethodOfObject(activity, in).isDefined
    }

    override def apply(in: AnyRef): AnyRef = {
      val method = findMethodOfObject(activity, in).get
      method.invoke(in)
    }
  }
}

// 2011-11-26
class G3AgentClass(val activity: Class[_], context: G3Context) extends G3Pipe(context) with ActivityActionable {
  import UAgent._
  protected def create_Activity(ctx: G3AgentContext) = new PartialFunction[AnyRef, AnyRef] {
    override def isDefinedAt(in: AnyRef): Boolean = {
      findMethodOfClass(activity, in).isDefined
    }

    override def apply(in: AnyRef): AnyRef = {
      val method = findMethodOfClass(activity, in).get
      method.invoke(in)
    }
  }
}

// 2011-11-28
class G3AgentOr(context: G3Context, val activities: AnyRef*) extends G3Pipe(context) with ActivityActionable {
  import UAgent._
  private val _actions: List[G3Action] = makeActions(context, activities.toList) 

  protected def create_Activity(ctx: G3AgentContext) = new PartialFunction[AnyRef, AnyRef] {
    override def isDefinedAt(in: AnyRef): Boolean = {
      _actions.exists(_.isDefinedAt(in))
    }

    override def apply(in: AnyRef): AnyRef = {
      _actions.find(_.isDefinedAt(in)) match {
        case Some(action) => action.apply(ctx, in)
        case None => in
      }
    }
  }
}

// 2011-11-26
trait UAgent {
  def findMethodOfObject(o: AnyRef, in: AnyRef): Option[Method] = {
    findMethodOfClass(o.getClass, in)
  }

  def findMethodOfClass(c: Class[_], in: AnyRef): Option[Method] = {
    val methods = c.getMethods()    
    val method: Option[Method] = methods find { m =>
      "apply".equals(m.getName()) && isValidMethod(m, in)
    } orElse { 
      methods find (isValidMethod(_, in))
    } 
    method
  }

  def isValidMethod(m: Method, in: AnyRef): Boolean = {
    val pts = m.getParameterTypes()
    pts.length == 1 &&
    pts(0).isAssignableFrom(in.getClass)
  }

  def isValidFunction(f: Function1[_, _], in: AnyRef): Boolean = {
    f.getClass.getMethods.exists { m =>
      "apply".equals(m.getName) &&
      isValidMethod(m, in)
    }
  }

  def makeActions(context: G3Context, activities: List[AnyRef]): List[G3Action] = {
    activities.map(makeAction(context, _))
  }

  def makeAction(context: G3Context, activity: AnyRef): G3Action = {
    activity match {
      case pf: PartialFunction[_, _] => new PartialFunctionAction(pf)
      case f: Function1[_, _] => new FunctionAction(f)
      case c: Class[_] => new ClassAction(c)
      case gp: GApplicationDescriptor => new GoldenportAction(context, gp) 
      case o: AnyRef => new ObjectAction(o)
    }
  }
}

object UAgent extends UAgent


class G3Filter(val activity: AnyRef => Boolean, context: G3Context) extends G3Pipe(context) {
  // XXX
}

class G3Map(val activity: AnyRef => AnyRef, context: G3Context) extends G3Pipe(context) {
  // XXX
}

class G3Invoke(val channel: String, context: G3Context) extends G3Pipe(context) {
  override def do_Action(parcel: G3Parcel): G3Parcel = {
//    println("G3Invoke.doAction")
    context.invoke(channel, parcel)
  }
}

class G3Publish(val subscribers: List[String], context: G3Context) extends G3Pipe(context) {
  override def do_Action(parcel: G3Parcel): G3Parcel = {
//    println("G3Publish.doAction")
    context.sendAll(subscribers, parcel)._2
  }
}

class G3Join[A](context: G3Context) extends G3Pipe(context) {
  override protected def pipe_execute(parcel: G3Parcel) {
    val rp = execute_action(parcel)
    val results = (for (ticket <- rp.tickets) yield {
      context.messageStore.getResultParcel(ticket)
    }).flatMap(p => p)
    log_debug("join results = " + results)
    if (true) {
      join_per_ticket(results, rp)
    } else {
      join_aggregate_tickets(results, rp)
    }
  }

  private def join_per_ticket(results: List[G3Parcel], source: G3Parcel) {
    for (result <- results) {
      pipe_next(context.createContentParcel("join")(result.content, source))
    }
  }

  private def join_aggregate_tickets(results: List[G3Parcel], source: G3Parcel) {
    pipe_next(context.createContentParcel("join")(results.map(_.content), source))
  }

/*
  //override
  def doAction0(parcel: G3Parcel): G3Parcel = {
    println("join doAction = " + parcel + "/" + parcel.tickets)
    val results = (for (ticket <- parcel.tickets) yield {
      val x = context.messageStore.getResultParcel(ticket)
      println("join result = " + x)
      x
    }) // .flatMap(e => e).toList
    println("join results = " + results)
    parcel.context.createContentAttachmentsParcel("join")(results, parcel.attachments, parcel)
  }
*/
}

class G3Split[A](val strategy: SplitStrategy, context: G3Context) extends G3Pipe(context) {
  private var counter = 0 // XXX single thread implementation

  override protected def pipe_execute(parcel: G3Parcel) {
    val rp = execute_action(parcel)
    rp.content match {
      case contents: List[AnyRef] if (contents.length > 1) => {
        counter = contents.length // XXX single thread implementation
        for (p <- context.createSplitParcels("split")(contents, rp)) {
//          println("Split:" + p + "/" + p.parent)
          pipe_next(p)
          counter -= 1 // XXX single thread implementation
        }
      }
      case _ => pipe_next(rp)
    }
  }

  override protected def is_Processing(uow: G3UnitofworkId) = {
    counter > 1 // XXX single thread implementation
  }
}

class G3Aggregate[A](context: G3Context) extends G3Pipe(context) {
  private val parcel_queue = new ArrayBuffer[AggregateHolder]

  private def is_partial(parcel: G3Parcel) = {
    parcel.parent.isDefined || !parcel.children.isEmpty
  }

  override protected def pipe_execute(parcel: G3Parcel) {
    val rp = execute_action(parcel)
    if (is_partial(parcel)) {
      aggregate_parts(rp) match {
        case Some(aggregated) => pipe_next(aggregated)
        case None => {}
      }
    } else {
      pipe_next(rp)
    }
  }

  private def aggregate_parts(parcel: G3Parcel): Option[G3Parcel] = {
    for (holder <- parcel_queue) {
//      println("aggregate_parts1: " + parcel + "/" + parcel.parent)
      if (holder.tryAddParcel(parcel)) {
//        println("aggregate_parts2: " + parcel)
        if (holder.isFilled) {
//          println("aggregate_parts3: " + parcel)
          parcel_queue -= holder
          val (parent, children) = holder.assemblyParts
          return Some(aggregate_parcels(parent, children))
        }
      }
    }
//    println("aggregate_parts4: " + parcel)
    val newHolder = new AggregateHolder()
    newHolder.initParcel(parcel)
    parcel_queue += newHolder
    return None
  }

  private def aggregate_parcels(parent: G3Parcel, children: List[G3Parcel]): G3Parcel  = {
    val sources: List[G3Parcel] = parent :: children
    val (content: AnyRef, attachments: List[AnyRef]) = aggregate_Parcels(sources)
    context.createAggregatedParcel("aggregate")(content, attachments, sources)
  }

  protected def aggregate_Parcels(sources: List[G3Parcel]): (AnyRef, List[AnyRef]) = {
    (sources.map(_.content), sources(0).attachments)
  }
}

// @seince  May.  3. 2010
// @version May.  5. 2010
class AggregateHolder() {
  private var target: Option[G3Parcel] = None
  private var children = new ArrayBuffer[Option[AggregateHolder]]

  private def assert_invariants {
    require (target.isDefined || children.length > 0)
    require ((target.isDefined && children.length == target.get.children.length) ||
             !target.isDefined)
    assert_invariants_heavy
  }

  private def assert_invariants_heavy {
  }

  def isFilled = {
    assert_invariants
    target.isDefined && !children.exists(_.isEmpty)
  }

  def childrenLength = target.flatMap(p => Some(p.children.length))

  def children_length = children.length

  def targetId: G3ParcelId = {
    assert_invariants
    if (target.isDefined) {
      val parcel = target.get
      if (parcel.base.isDefined) parcel.base.get
      else parcel.id
    }
    else {
      children.find(h => h.isDefined) match {
        case Some(Some(holder)) => holder.target.get.parent.get
        case _ => error("XXX")
      }
    }
  }

  def initParcel(parcel: G3Parcel) = {
    if (parcel.parent.isDefined) {
      put_child(parcel)
    } else {
      put_target(parcel)
    }
    this
  }

  def initTargetParcel(parcel: G3Parcel) = {
    put_target(parcel)
    this
  }

  def tryAddParcel(parcel: G3Parcel): Boolean = {
    val xx = if (parcel.parent.isDefined) {
      parcel.parent.get
    }  else {
      "None"
    }
//    println("tryAddParcel: %s, %s".format(xx, targetId))
    if (parcel.base.isDefined && parcel.base.get == targetId) {
      put_target(parcel)
      return true
    } else if (parcel.parent.isDefined && parcel.parent.get == targetId) {
      require (parcel.sequenceNo > 0)
      put_child(parcel)
      return true
    } else {
      for (child <- children) {
        if (child.isDefined && child.get.tryAddParcel(parcel)) {
          return true
        }
      }
      return false
    }
  }

  private def put_target(parcel: G3Parcel) {
    assume (target.isEmpty) // ??? idempotent
    target = Some(parcel)
    ensure_children_space(parcel.children.length)
  }

  private def put_child(parcel: G3Parcel) {
    val seqNo = parcel.sequenceNo
    require (seqNo > 0)
    val childIndex = seqNo - 1
    ensure_children_space(childIndex + 1)
    assume (children(childIndex).isEmpty)
    val holder = new AggregateHolder()
    holder.target = Some(parcel)
    children(childIndex) =  Some(holder)
  }

  private def ensure_children_space(size: Int) {
    if (children.length >= size) return
    for (i <- children.length until size) {
      children += None
    }
  }

  def assemblyParts = (target.get, children.map(_.get.target.get).toList)
}
